"""
Case Type   : D库常用语法兼容-PART2
Case Name   : pdbc驱动测试
Create At   : 2025/10/17
Owner       : @chen
Description :
    1.解压openGauss-*-CentOS-x86_64-Python.tar.gz得到lib,替换psycopg2
    2.配置白名单
    3.配置lib
    4.连接数据库
    5.ansi_nulls功能验证
    6.rotate功能验证
    7.断开连接
Expect      :
    1.拷贝成功
    2.配置成功
    3.配置成功
    4.连接数据库成功
    5.执行成功
    6.执行成功
    7.执行成功
History     :
"""

import os
import unittest

from testcase.CONNECTORS.PDBC.PSYCOPG2. \
    Opengauss_Function_Connect_Psycopg_Common import Psycopg2Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger

primary_sh = CommonSH('PrimaryDbUser')

class Not_Rotate0036(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        text = f'-----{os.path.basename(__file__)} start-----'
        self.log.info(text)
        self.pri_sh = CommonSH('PrimaryDbDmode')
        self.com = Common()
        self.constant = Constant()
        self.conn_info = {'dbname': self.pri_sh.node.db_name,
                          'user': self.pri_sh.node.db_user,
                          'password': self.pri_sh.node.db_password,
                          'host': self.pri_sh.node.ssh_host,
                          'port': self.pri_sh.node.db_port}
        self.psycopg_com = Psycopg2Common(**self.conn_info)
        self.ansi_param = self.com.show_param('ansi_nulls')

        text = '----前置操作: 备份白名单 expect: 备份成功----'
        self.log.info(text)
        res = self.psycopg_com.bak_conf(bak_flag=True)
        self.log.info(res)
        self.assertTrue(res, f'执行失败: {text}')

    def test_not_rotate(self):
        text = '----step1: 解压PDBC得到lib,替换psycopg2 expect: 执行成功----'
        res = self.psycopg_com.decompress_pkg()
        self.assertTrue(res, f'执行失败: {text}')
        res = self.psycopg_com.set_new_psycopg2()
        self.assertTrue(res, f'执行失败: {text}')

        text = '----step2: 配置白名单 expect: 配置成功----'
        self.log.info(text)
        res = self.psycopg_com.set_pghba(open_flag=True)
        self.assertTrue(res, f'执行失败: {text}')

        text = '----step3: 配置lib expect: 成功----'
        self.log.info(text)
        res = self.psycopg_com.set_lib_path()
        self.assertTrue(res, f'执行失败: {text}')

        text = '----step4: 连接数据库 expect: 成功----'
        self.log.info(text)
        res = self.psycopg_com.conn_db()
        expect = f"user={self.pri_sh.node.db_user} " \
            f"dbname={self.pri_sh.node.db_name} " \
            f"host={self.pri_sh.node.ssh_host} " \
            f"port={self.pri_sh.node.db_port}"
        res = [self.assertIn(i, res) for i in expect.split()]
        self.assertEqual(len(res), 4, f'执行失败: {text}')

        text = '----step5: ansi_nulls功能验证 expect: 成功----'
        self.log.info(text)
        f_sql = f'cur.execute(\\\"show ansi_nulls;\\\")\n' \
            f'cur.execute(\\\"set ansi_nulls on;\\\")\n' \
            f'cur.execute(\\\"select null =null;\\\")\n' \
            f'cur.execute(\\\"set ansi_nulls off;\\\")\n' \
            f'cur.execute(\\\"select null =null;\\\")\n'
        run_sql = self.psycopg_com.join_run_sql(f_sql)
        res = self.psycopg_com.run_sql(sql=run_sql)
        self.assertIn('[(True,)]', res)

        text = '----step6: rotate功能验证 expect: 成功----'
        self.log.info(text)
        f_sql = f'cur.execute(\\\"drop table if exists t_rotate0001;\\\")\n' \
            f'cur.execute(\\\"CREATE TABLE t_rotate0001 (Product VARCHAR(50)' \
            f',Quarter CHAR(2),sales DECIMAL(10,2));\\\")\n' \
            f"cur.execute(\\\"INSERT INTO t_rotate0001 VALUES('Laptop', 'Q1'," \
            f" 1000.00),('Laptop', 'Q2', 1500.00),('Phone', 'Q1', 800.00)," \
            f"('Phone', 'Q3', 2000.00),('Tablet', 'Q4', 500.00);\\\")\n" \
            f'cur.execute(\\\"SELECT Product, Q1, Q2, Q3, Q4 FROM' \
            f' (SELECT Product, Quarter, sales FROM t_rotate0001) AS' \
            f' SourceTable rotate (SUM(sales) FOR Quarter IN (Q1, Q2, Q3,' \
            f' Q4)) AS rotateTable;\\\")\n'
        run_sql = self.psycopg_com.join_run_sql(f_sql)
        res = self.psycopg_com.run_sql(sql=run_sql)
        self.assertIn(r"[('Tablet', None, None, None, Decimal('500.00'))," \
            r" ('Laptop', Decimal('1000.00'), Decimal('1500.00')," \
            r" None, None), ('Phone', Decimal('800.00'), None, " \
            r"Decimal('2000.00'), None)]", res)

    def tearDown(self):
        text = '----run teardown----'
        self.log.info(text)
        f_sql = f'cur.execute(\\\"drop table if exists t_rotate0001;\\\")\n'
        self.log.info(f_sql)
        run_sql = self.psycopg_com.join_run_sql(f_sql)
        res = self.psycopg_com.run_sql(sql=run_sql)
        set_param = primary_sh.execute_gsguc('reload',
                                    self.constant.GSGUC_SUCCESS_MSG,
                                    f"ansi_nulls = {self.ansi_param}")
        self.log.info(set_param)
        self.assertTrue(set_param, '参数修改异常')
        text1 = f'----后置操作：删除解压文件目录----'
        self.log.info(text1)
        res1 = self.psycopg_com.clean_decompress_file()

        text2 = f'----后置操作：还原白名单----'
        self.log.info(text2)
        res2 = self.psycopg_com.bak_conf(bak_flag=False)

        self.assertTrue(res1, f'执行失败: {text1}')
        self.assertTrue(res2, f'执行失败: {text2}')